/**
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package com.topsoft.botspider.io;

import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.WeakHashMap;

import org.apache.avro.AvroRuntimeException;
import org.apache.avro.Schema;
import org.apache.avro.Schema.Field;
import org.apache.avro.Schema.Type;
import org.apache.avro.generic.GenericContainer;
import org.apache.avro.generic.GenericData;
import org.apache.avro.generic.IndexedRecord;
import org.apache.avro.specific.SpecificData.SchemaConstructable;

/** A key/value pair. */
public class Pair<K, V> implements IndexedRecord, Comparable<Pair>,
		SchemaConstructable {

	private static final String PAIR = Pair.class.getName();
	private static final String KEY = "key";
	private static final String VALUE = "value";

	private Schema schema;
	private K key;
	private V value;

	public Pair(Schema schema) {
		checkIsPairSchema(schema);
		this.schema = schema;
	}

	public void setSchema(Schema schema) {
		checkIsPairSchema(schema);
		this.schema = schema;
	}

	public Pair(K key, Schema keySchema, V value, Schema valueSchema) {
		this.schema = getPairSchema(keySchema, valueSchema);
		this.key = key;
		this.value = value;
	}

	private static void checkIsPairSchema(Schema schema) {
	  return;
//    if (!PAIR.equals(schema.getFullName()))
//      throw new IllegalArgumentException("Not a Pair schema: "+schema);
  }

	/** Return a pair's key schema. */
	public static Schema getKeySchema(Schema pair) {
		checkIsPairSchema(pair);
		return pair.getField(KEY).schema();
	}

	/** Return a pair's value schema. */
	public static Schema getValueSchema(Schema pair) {
		checkIsPairSchema(pair);
		return pair.getField(VALUE).schema();
	}

	private static final Map<Schema, Map<Schema, Schema>> SCHEMA_CACHE = new WeakHashMap<Schema, Map<Schema, Schema>>();

	/** Get a pair schema. */
	public static Schema getPairSchema(Schema key, Schema value) {
		Map<Schema, Schema> valueSchemas;
		synchronized (SCHEMA_CACHE) {
			valueSchemas = SCHEMA_CACHE.get(key);
			if (valueSchemas == null) {
				valueSchemas = new WeakHashMap<Schema, Schema>();
				SCHEMA_CACHE.put(key, valueSchemas);
			}
			Schema result;
			result = valueSchemas.get(value);
			if (result == null) {
				result = makePairSchema(key, value);
				valueSchemas.put(value, result);
			}
			return result;
		}
	}

	private static Schema makePairSchema(Schema key, Schema value) {
		Schema pair = Schema.createRecord(PAIR, null, null, false);
		List<Field> fields = new ArrayList<Field>();
		fields.add(new Field(KEY, key, "", null));
		fields.add(new Field(VALUE, value, "", null, Field.Order.IGNORE));
		pair.setFields(fields);
		return pair;
	}

	@Override
	public Schema getSchema() {
		return schema;
	}

	/** Get the key. */
	public K key() {
		return key;
	}

	/** Set the key. */
	public void key(K key) {
		this.key = key;
	}

	/** Get the value. */
	public V value() {
		return value;
	}

	/** Set the value. */
	public void value(V value) {
		this.value = value;
	}

	/** Set both the key and value. */
	public void set(K key, V value) {
		this.key = key;
		this.value = value;
	}

	@Override
	public boolean equals(Object o) {
		if (o == this)
			return true; // identical object
		if (!(o instanceof Pair))
			return false; // not a pair
		Pair that = (Pair) o;
		if (this.schema != that.schema)
			return false; // not the same schema
		return this.compareTo(that) == 0;
	}

	@Override
	public int hashCode() {
		return GenericData.get().hashCode(this, schema);
	}

	@Override
	public int compareTo(Pair that) {
		return GenericData.get().compare(this, that, schema);
	}

	@Override
	public String toString() {
		return GenericData.get().toString(this);
	}

	@Override
	public Object get(int i) {
		switch (i) {
		case 0:
			return key;
		case 1:
			return value;
		default:
			throw new AvroRuntimeException("Bad index: " + i);
		}
	}

	@Override
	@SuppressWarnings("unchecked")
	public void put(int i, Object o) {
		switch (i) {
		case 0:
			this.key = (K) o;
			break;
		case 1:
			this.value = (V) o;
			break;
		default:
			throw new AvroRuntimeException("Bad index: " + i);
		}
	}

	private static final Schema STRING_SCHEMA = Schema.create(Type.STRING);
	private static final Schema BYTES_SCHEMA = Schema.create(Type.BYTES);
	private static final Schema INT_SCHEMA = Schema.create(Type.INT);
	private static final Schema LONG_SCHEMA = Schema.create(Type.LONG);
	private static final Schema FLOAT_SCHEMA = Schema.create(Type.FLOAT);
	private static final Schema DOUBLE_SCHEMA = Schema.create(Type.DOUBLE);
	private static final Schema NULL_SCHEMA = Schema.create(Type.NULL);

	@SuppressWarnings("unchecked")
	public Pair(GenericContainer key, GenericContainer value) {
		this((K) key, key.getSchema(), (V) value, value.getSchema());
	}

	@SuppressWarnings("unchecked")
	public Pair(GenericContainer key, CharSequence value) {
		this((K) key, key.getSchema(), (V) value, STRING_SCHEMA);
	}

	@SuppressWarnings("unchecked")
	public Pair(GenericContainer key, ByteBuffer value) {
		this((K) key, key.getSchema(), (V) value, BYTES_SCHEMA);
	}

	@SuppressWarnings("unchecked")
	public Pair(GenericContainer key, Integer value) {
		this((K) key, key.getSchema(), (V) value, INT_SCHEMA);
	}

	@SuppressWarnings("unchecked")
	public Pair(GenericContainer key, Long value) {
		this((K) key, key.getSchema(), (V) value, LONG_SCHEMA);
	}

	@SuppressWarnings("unchecked")
	public Pair(GenericContainer key, Float value) {
		this((K) key, key.getSchema(), (V) value, FLOAT_SCHEMA);
	}

	@SuppressWarnings("unchecked")
	public Pair(GenericContainer key, Double value) {
		this((K) key, key.getSchema(), (V) value, DOUBLE_SCHEMA);
	}

	@SuppressWarnings("unchecked")
	public Pair(GenericContainer key, Void value) {
		this((K) key, key.getSchema(), (V) value, NULL_SCHEMA);
	}

	@SuppressWarnings("unchecked")
	public Pair(CharSequence key, GenericContainer value) {
		this((K) key, STRING_SCHEMA, (V) value, value.getSchema());
	}

	@SuppressWarnings("unchecked")
	public Pair(CharSequence key, CharSequence value) {
		this((K) key, STRING_SCHEMA, (V) value, STRING_SCHEMA);
	}

	@SuppressWarnings("unchecked")
	public Pair(CharSequence key, ByteBuffer value) {
		this((K) key, STRING_SCHEMA, (V) value, BYTES_SCHEMA);
	}

	@SuppressWarnings("unchecked")
	public Pair(CharSequence key, Integer value) {
		this((K) key, STRING_SCHEMA, (V) value, INT_SCHEMA);
	}

	@SuppressWarnings("unchecked")
	public Pair(CharSequence key, Long value) {
		this((K) key, STRING_SCHEMA, (V) value, LONG_SCHEMA);
	}

	@SuppressWarnings("unchecked")
	public Pair(CharSequence key, Float value) {
		this((K) key, STRING_SCHEMA, (V) value, FLOAT_SCHEMA);
	}

	@SuppressWarnings("unchecked")
	public Pair(CharSequence key, Double value) {
		this((K) key, STRING_SCHEMA, (V) value, DOUBLE_SCHEMA);
	}

	@SuppressWarnings("unchecked")
	public Pair(CharSequence key, Void value) {
		this((K) key, STRING_SCHEMA, (V) value, NULL_SCHEMA);
	}

	@SuppressWarnings("unchecked")
	public Pair(ByteBuffer key, GenericContainer value) {
		this((K) key, BYTES_SCHEMA, (V) value, value.getSchema());
	}

	@SuppressWarnings("unchecked")
	public Pair(ByteBuffer key, CharSequence value) {
		this((K) key, BYTES_SCHEMA, (V) value, STRING_SCHEMA);
	}

	@SuppressWarnings("unchecked")
	public Pair(ByteBuffer key, ByteBuffer value) {
		this((K) key, BYTES_SCHEMA, (V) value, BYTES_SCHEMA);
	}

	@SuppressWarnings("unchecked")
	public Pair(ByteBuffer key, Integer value) {
		this((K) key, BYTES_SCHEMA, (V) value, INT_SCHEMA);
	}

	@SuppressWarnings("unchecked")
	public Pair(ByteBuffer key, Long value) {
		this((K) key, BYTES_SCHEMA, (V) value, LONG_SCHEMA);
	}

	@SuppressWarnings("unchecked")
	public Pair(ByteBuffer key, Float value) {
		this((K) key, BYTES_SCHEMA, (V) value, FLOAT_SCHEMA);
	}

	@SuppressWarnings("unchecked")
	public Pair(ByteBuffer key, Double value) {
		this((K) key, BYTES_SCHEMA, (V) value, DOUBLE_SCHEMA);
	}

	@SuppressWarnings("unchecked")
	public Pair(ByteBuffer key, Void value) {
		this((K) key, BYTES_SCHEMA, (V) value, NULL_SCHEMA);
	}

	@SuppressWarnings("unchecked")
	public Pair(Integer key, GenericContainer value) {
		this((K) key, INT_SCHEMA, (V) value, value.getSchema());
	}

	@SuppressWarnings("unchecked")
	public Pair(Integer key, CharSequence value) {
		this((K) key, INT_SCHEMA, (V) value, STRING_SCHEMA);
	}

	@SuppressWarnings("unchecked")
	public Pair(Integer key, ByteBuffer value) {
		this((K) key, INT_SCHEMA, (V) value, BYTES_SCHEMA);
	}

	@SuppressWarnings("unchecked")
	public Pair(Integer key, Integer value) {
		this((K) key, INT_SCHEMA, (V) value, INT_SCHEMA);
	}

	@SuppressWarnings("unchecked")
	public Pair(Integer key, Long value) {
		this((K) key, INT_SCHEMA, (V) value, LONG_SCHEMA);
	}

	@SuppressWarnings("unchecked")
	public Pair(Integer key, Float value) {
		this((K) key, INT_SCHEMA, (V) value, FLOAT_SCHEMA);
	}

	@SuppressWarnings("unchecked")
	public Pair(Integer key, Double value) {
		this((K) key, INT_SCHEMA, (V) value, DOUBLE_SCHEMA);
	}

	@SuppressWarnings("unchecked")
	public Pair(Integer key, Void value) {
		this((K) key, INT_SCHEMA, (V) value, NULL_SCHEMA);
	}

	@SuppressWarnings("unchecked")
	public Pair(Long key, GenericContainer value) {
		this((K) key, LONG_SCHEMA, (V) value, value.getSchema());
	}

	@SuppressWarnings("unchecked")
	public Pair(Long key, CharSequence value) {
		this((K) key, LONG_SCHEMA, (V) value, STRING_SCHEMA);
	}

	@SuppressWarnings("unchecked")
	public Pair(Long key, ByteBuffer value) {
		this((K) key, LONG_SCHEMA, (V) value, BYTES_SCHEMA);
	}

	@SuppressWarnings("unchecked")
	public Pair(Long key, Integer value) {
		this((K) key, LONG_SCHEMA, (V) value, INT_SCHEMA);
	}

	@SuppressWarnings("unchecked")
	public Pair(Long key, Long value) {
		this((K) key, LONG_SCHEMA, (V) value, LONG_SCHEMA);
	}

	@SuppressWarnings("unchecked")
	public Pair(Long key, Float value) {
		this((K) key, LONG_SCHEMA, (V) value, FLOAT_SCHEMA);
	}

	@SuppressWarnings("unchecked")
	public Pair(Long key, Double value) {
		this((K) key, LONG_SCHEMA, (V) value, DOUBLE_SCHEMA);
	}

	@SuppressWarnings("unchecked")
	public Pair(Long key, Void value) {
		this((K) key, LONG_SCHEMA, (V) value, NULL_SCHEMA);
	}

	@SuppressWarnings("unchecked")
	public Pair(Float key, GenericContainer value) {
		this((K) key, FLOAT_SCHEMA, (V) value, value.getSchema());
	}

	@SuppressWarnings("unchecked")
	public Pair(Float key, CharSequence value) {
		this((K) key, FLOAT_SCHEMA, (V) value, STRING_SCHEMA);
	}

	@SuppressWarnings("unchecked")
	public Pair(Float key, ByteBuffer value) {
		this((K) key, FLOAT_SCHEMA, (V) value, BYTES_SCHEMA);
	}

	@SuppressWarnings("unchecked")
	public Pair(Float key, Integer value) {
		this((K) key, FLOAT_SCHEMA, (V) value, INT_SCHEMA);
	}

	@SuppressWarnings("unchecked")
	public Pair(Float key, Long value) {
		this((K) key, FLOAT_SCHEMA, (V) value, LONG_SCHEMA);
	}

	@SuppressWarnings("unchecked")
	public Pair(Float key, Float value) {
		this((K) key, FLOAT_SCHEMA, (V) value, FLOAT_SCHEMA);
	}

	@SuppressWarnings("unchecked")
	public Pair(Float key, Double value) {
		this((K) key, FLOAT_SCHEMA, (V) value, DOUBLE_SCHEMA);
	}

	@SuppressWarnings("unchecked")
	public Pair(Float key, Void value) {
		this((K) key, FLOAT_SCHEMA, (V) value, NULL_SCHEMA);
	}

	@SuppressWarnings("unchecked")
	public Pair(Double key, GenericContainer value) {
		this((K) key, DOUBLE_SCHEMA, (V) value, value.getSchema());
	}

	@SuppressWarnings("unchecked")
	public Pair(Double key, CharSequence value) {
		this((K) key, DOUBLE_SCHEMA, (V) value, STRING_SCHEMA);
	}

	@SuppressWarnings("unchecked")
	public Pair(Double key, ByteBuffer value) {
		this((K) key, DOUBLE_SCHEMA, (V) value, BYTES_SCHEMA);
	}

	@SuppressWarnings("unchecked")
	public Pair(Double key, Integer value) {
		this((K) key, DOUBLE_SCHEMA, (V) value, INT_SCHEMA);
	}

	@SuppressWarnings("unchecked")
	public Pair(Double key, Long value) {
		this((K) key, DOUBLE_SCHEMA, (V) value, LONG_SCHEMA);
	}

	@SuppressWarnings("unchecked")
	public Pair(Double key, Float value) {
		this((K) key, DOUBLE_SCHEMA, (V) value, FLOAT_SCHEMA);
	}

	@SuppressWarnings("unchecked")
	public Pair(Double key, Double value) {
		this((K) key, DOUBLE_SCHEMA, (V) value, DOUBLE_SCHEMA);
	}

	@SuppressWarnings("unchecked")
	public Pair(Double key, Void value) {
		this((K) key, DOUBLE_SCHEMA, (V) value, NULL_SCHEMA);
	}

	@SuppressWarnings("unchecked")
	public Pair(Void key, GenericContainer value) {
		this((K) key, NULL_SCHEMA, (V) value, value.getSchema());
	}

	@SuppressWarnings("unchecked")
	public Pair(Void key, CharSequence value) {
		this((K) key, NULL_SCHEMA, (V) value, STRING_SCHEMA);
	}

	@SuppressWarnings("unchecked")
	public Pair(Void key, ByteBuffer value) {
		this((K) key, NULL_SCHEMA, (V) value, BYTES_SCHEMA);
	}

	@SuppressWarnings("unchecked")
	public Pair(Void key, Integer value) {
		this((K) key, NULL_SCHEMA, (V) value, INT_SCHEMA);
	}

	@SuppressWarnings("unchecked")
	public Pair(Void key, Long value) {
		this((K) key, NULL_SCHEMA, (V) value, LONG_SCHEMA);
	}

	@SuppressWarnings("unchecked")
	public Pair(Void key, Float value) {
		this((K) key, NULL_SCHEMA, (V) value, FLOAT_SCHEMA);
	}

	@SuppressWarnings("unchecked")
	public Pair(Void key, Double value) {
		this((K) key, NULL_SCHEMA, (V) value, DOUBLE_SCHEMA);
	}

	@SuppressWarnings("unchecked")
	public Pair(Void key, Void value) {
		this((K) key, NULL_SCHEMA, (V) value, NULL_SCHEMA);
	}

	// private static final String[][] TABLE = new String[][] {
	// {"GenericContainer", "{0}.getSchema()"},
	// {"CharSequence", "STRING_SCHEMA"},
	// {"ByteBuffer", "BYTES_SCHEMA"},
	// {"Integer", "INT_SCHEMA"},
	// {"Long", "LONG_SCHEMA"},
	// {"Float", "FLOAT_SCHEMA"},
	// {"Double", "DOUBLE_SCHEMA"},
	// {"Void", "NULL_SCHEMA"},
	// };

	// private static String f(String pattern, String value) {
	// return java.text.MessageFormat.format(pattern, value);
	// }

	// public static void main(String... args) throws Exception {
	// StringBuffer b = new StringBuffer();
	// for (String[] k : TABLE) {
	// for (String[] v : TABLE) {
	// b.append("@SuppressWarnings(\"unchecked\")\n");
	// b.append("public Pair("+k[0]+" key, "+v[0]+" value) {\n");
	// b.append("  this((K)key, "+f(k[1],"key")
	// +", (V)value, "+f(v[1],"value")+");\n");
	// b.append("}\n");
	// }
	// }
	// System.out.println(b);
	// }

}
